package net.pantolomin.nayo.test.perf;

import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import net.pantolomin.nayo.NioClient;
import net.pantolomin.nayo.NioProvider;
import net.pantolomin.nayo.NioServer;
import net.pantolomin.nayo.api.Connection;
import net.pantolomin.nayo.api.ConnectionEventsListener;

/**
 * A node in the network
 */
public class Node implements ConnectionEventsListener {
	/** The provider */
	private final NioProvider provider;

	/** The listening port */
	private final int listeningPort;

	/** The listener */
	private final PerfTest listener;

	/** The nodes (including us) */
	private final Map<Integer, RemoteNode> nodes;

	/** The server */
	private NioServer server;

	/**
	 * Create a node
	 * 
	 * @param provider
	 *          the provider
	 * @param listeningPort
	 *          the listening port
	 * @param nodes
	 *          the nodes (including us)
	 * @param listener
	 *          the listener
	 */
	public Node(NioProvider provider, int listeningPort, int[] nodes, PerfTest listener) {
		this.provider = provider;
		this.listeningPort = listeningPort;
		this.listener = listener;
		this.nodes = new HashMap<Integer, RemoteNode>(nodes.length);
		for (int node : nodes) {
			if (node != listeningPort) {
				this.nodes.put(Integer.valueOf(node), new RemoteNode());
			}
		}
	}

	/**
	 * Start the node
	 * 
	 * @throws IOException
	 *           if an error occurred
	 */
	public void start() throws IOException {
		this.server = this.provider.listen(InetAddress.getLocalHost(), this.listeningPort, null, this);
	}

	/**
	 * Connect the node to the network
	 * 
	 * @throws IOException
	 *           if an error occurred
	 */
	public void connect() throws IOException {
		for (Entry<Integer, RemoteNode> node : this.nodes.entrySet()) {
			NioClient client = this.provider.connect(InetAddress.getLocalHost(), node.getKey().intValue(), null, this);
			node.getValue().setClient(client);
		}
	}

	/**
	 * Disconnect the node from the network
	 */
	public void disconnect() {
		for (RemoteNode node : this.nodes.values()) {
			if (node.getClient() != null) {
				node.getClient().stop();
			}
		}
	}

	/**
	 * Stop the node
	 */
	public void stop() {
		if (this.server != null) {
			this.server.stop();
		}
	}

	//
	//
	//
	//
	//
	//
	//
	//
	//
	//
	// ************************************************************************
	// ************************************************************************
	// LISTENER METHODS
	// ************************************************************************
	// ************************************************************************

	public void newConnection(Connection connection) {
		try {
			int remotePort = connection.getRemoteAddress().getPort();
			RemoteNode remoteNode = this.nodes.get(Integer.valueOf(remotePort));
			if (remoteNode != null) {
				remoteNode.setConnection(connection);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		this.listener.newConnection(connection);
	}

	public void bindDone(InetAddress hostAddress, int port) {
		this.listener.bindDone(hostAddress, port);
	}

	public void bindFailed(InetAddress hostAddress, int port, String cause) {
		this.listener.bindFailed(hostAddress, port, cause);
	}

	public void connectionTimeout(InetAddress hostAddress, int port) {
		this.listener.connectionTimeout(hostAddress, port);
	}

	public void connectionFailed(InetAddress hostAddress, int port, String cause) {
		this.listener.connectionFailed(hostAddress, port, cause);
	}

	public void connectionClosed(Connection connection) {
		this.listener.connectionClosed(connection);
	}

	public void closedByRemote(Connection connection, boolean cleanly) {
		this.listener.closedByRemote(connection, cleanly);
	}

	private final Map<Connection, ByteBuffer> readBuffers = new ConcurrentHashMap<Connection, ByteBuffer>();

	public void dataRead(Connection connection, byte[] data) {
		if (connection != null) {
			ByteBuffer readBuffer = readBuffers.get(connection);
			if (readBuffer == null) {
				readBuffer = ByteBuffer.allocate(102400);
				readBuffers.put(connection, readBuffer);
			}
			readBuffer.put(data);
			readBuffer.flip();
			while (readBuffer.hasRemaining()) {
				readBuffer.mark();
				Message msg = new Message(readBuffer);
				if (msg.getLength() == 0) {
					readBuffer.reset();
					break;
				}
				if (!this.listener.declareMessage(msg)) {
					forwardMessage(msg);
				}
			}
			if (readBuffer.hasRemaining()) {
				readBuffer.compact();
			} else {
				readBuffer.clear();
			}
		} else {
			forwardMessage(new Message(ByteBuffer.wrap(data).getInt()));
		}
	}

	/**
	 * Forward a message
	 * 
	 * @param msg
	 *          the message
	 */
	private void forwardMessage(Message msg) {
		// Forward the message to a random node
		int index = (int) Math.floor(Math.random() * this.nodes.size());
		for (RemoteNode node : this.nodes.values()) {
			if (index == 0) {
				ByteBuffer writeBuffer = ByteBuffer.allocateDirect(8 * (msg.getLength() + 2));
				msg.writeMessage(writeBuffer);
				writeBuffer.flip();
				try {
					node.getConnection().send(writeBuffer);
				} catch (IOException e) {
					this.listener.writeFailed(node.getConnection(), e);
				}
				return;
			}
			index--;
		}
	}

	public void writeFailed(Connection connection, IOException exception) {
		this.listener.writeFailed(connection, exception);
	}

	public void writeBufferFull(Connection connection) {
		this.listener.writeBufferFull(connection);
	}

	public void writeBufferFreed(Connection connection) {
		this.listener.writeBufferFreed(connection);
	}

	//
	//
	//
	//
	//
	//
	//
	//
	//
	//
	// ************************************************************************
	// ************************************************************************
	// REMOTE NODE CONTEXT
	// ************************************************************************
	// ************************************************************************

	/**
	 * Represents a remote node
	 */
	private static final class RemoteNode {
		/** The client */
		private NioClient client;

		/** The connection */
		private Connection connection;

		public NioClient getClient() {
			return client;
		}

		public void setClient(NioClient client) {
			this.client = client;
		}

		public Connection getConnection() {
			return connection;
		}

		public void setConnection(Connection connection) {
			this.connection = connection;
		}
	}
}